package com.binance.notification.web.service.abnormaltrade.impl;

import java.math.BigDecimal;
import java.math.MathContext;
import java.math.RoundingMode;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StopWatch;

import com.binance.notification.data.mapper.AbnormalThresholdConfigMapper;
import com.binance.notification.data.models.AbnormalThresholdConfigModel;
import com.binance.notification.web.service.abnormaltrade.IAbnormalPriceService;
import com.binance.notification.web.service.abnormaltrade.IAbnormalTradeNoticeLogService;
import com.binance.notification.web.service.abnormaltrade.IAbnormalTradePushNoticeService;
import com.binance.notification.web.service.abnormaltrade.convert.AbnormalPricePushMessageConvertor;
import com.binance.notification.web.service.abnormaltrade.enums.AbnormalTradePushNoticeBusTypeEnum;
import com.binance.notification.web.service.abnormaltrade.enums.AbnromalTradeBusinessTypeEnum;
import com.binance.notification.web.service.abnormaltrade.enums.AbnromalTradeNoticeDescEnum;
import com.binance.notification.web.service.abnormaltrade.enums.IntervalEnum;
import com.binance.notification.web.service.abnormaltrade.integration.KlineService;
import com.binance.notification.web.service.abnormaltrade.log.LoggerUtils;
import com.binance.notification.web.service.abnormaltrade.models.AbnormalTradeNoticeInfo;
import com.binance.notification.web.service.abnormaltrade.models.AbnormalTradeNoticeLogBo;
import com.binance.notification.web.service.abnormaltrade.models.KlineInfo;
import com.binance.notification.web.utils.Worker;
import com.binance.sysconf.service.SysConfigVarCacheService;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;

import io.shardingsphere.api.HintManager;

@Service
public class AbnormalPriceServiceImpl implements IAbnormalPriceService {
    private static final org.slf4j.Logger log = LoggerUtils.getLogger(AbnormalPriceServiceImpl.class);
    @Autowired
    private KlineService klineService;
    @Autowired
    private AbnormalThresholdConfigMapper abnormalThresholdConfigMapper;
    @Autowired
    private IAbnormalTradeNoticeLogService abnormalTradeNoticeLogService;
    @Autowired
    private SysConfigVarCacheService sysConfigService;
    @Autowired
    private IAbnormalTradePushNoticeService abnormalTradePushNoticeService;
    @Autowired
    private Worker worker;
    @Override
    public void generateUsdtAbnormalTradeNotice(Set<String> symbolSet) {
        monitorSymbolListAndStore(symbolSet);
    }

    @Override
    public void generateBtcAbnormalTradeNotice(Set<String> symbolSet) {
        monitorSymbolListAndStore(symbolSet);
    }

    // @Override
    // public List<AbnormalTradeNoticeLogBo> getAbnormalTradeNoticeInfoList(Integer page, Integer
    // offset) {
    // List<AbnormalTradeNoticeLogBo> abnormalTradeNoticeLogBoList = abnormalTradeNoticeLogService
    // .findByBusinessType(AbnromalTradeBusinessTypeEnum.AbnormalPrice.name(), page, offset);
    // List<AbnormalTradeNoticeLogBo> wrapAbnormalTradeNoticeLogBoList =
    // wrapNeedPushFlag(abnormalTradeNoticeLogBoList);
    // return wrapAbnormalTradeNoticeLogBoList;
    // }

    /**
     * 格式化数据判断是否需要推送
     */
    // @Override
    // public List<AbnormalTradeNoticeLogBo> wrapNeedPushFlag(
    // List<AbnormalTradeNoticeLogBo> abnormalTradeNoticeLogBoList) {
    // List<AbnormalTradeNoticeLogBo> resultList = Lists.newArrayList();
    // log.info("AbnormalPriceServiceImpl.wrapNeedPushFlag strat abnormalTradeNoticeLogBoList={}",
    // abnormalTradeNoticeLogBoList);
    // for (AbnormalTradeNoticeLogBo abnormalTradeNoticeLogBo : abnormalTradeNoticeLogBoList) {
    // try {
    // log.info("AbnormalPriceServiceImpl.wrapNeedPushFlag abnormalTradeNoticeLogBo={}",
    // abnormalTradeNoticeLogBo);
    // Map<String, String> params = abnormalTradeNoticeLogBo.getBusinessData();
    // BigDecimal priceChangePercent = new BigDecimal(params.get("priceChangePercent"));
    // // 需要再覆盖一次
    // Boolean pushFlag = convertPushFlag(priceChangePercent, abnormalTradeNoticeLogBo.getSymbol());
    // params.put("pushFlag", String.valueOf(pushFlag));
    // abnormalTradeNoticeLogBo.setBusinessData(params);
    // log.info("AbnormalPriceServiceImpl.wrapNeedPushFlag singlefinish
    // abnormalTradeNoticeLogBo={}",
    // abnormalTradeNoticeLogBo);
    // resultList.add(abnormalTradeNoticeLogBo);
    // } catch (Exception e) {
    // log.error("AbnormalPriceServiceImpl.wrapNeedPushFlag Exception=", e);
    // continue;
    // }
    // }
    // log.info("AbnormalPriceServiceImpl.wrapNeedPushFlag finish resultList={}", resultList);
    // return resultList;
    // }

    /**
     * 格式化数据判断是否需要推送
     */
    @Override
    public AbnormalTradeNoticeLogBo wrapNeedPushFlag(AbnormalTradeNoticeLogBo bo) {
        try {
            log.info("AbnormalPriceServiceImpl.wrapNeedPushFlag abnormalTradeNoticeLogBo={}", bo);
            Map<String, String> params = bo.getBusinessData();
            BigDecimal priceChangePercent = new BigDecimal(params.get("priceChangePercent"));
            // 需要再覆盖一次
            Boolean pushFlag = convertPushFlag(priceChangePercent, bo.getSymbol());
            params.put("pushFlag", String.valueOf(pushFlag));
            bo.setBusinessData(params);
            log.info("AbnormalPriceServiceImpl.wrapNeedPushFlag singlefinish abnormalTradeNoticeLogBo={}", bo);
        } catch (Exception e) {
            log.error("AbnormalPriceServiceImpl.wrapNeedPushFlag Exception=", e);
        }
        log.info("AbnormalPriceServiceImpl.wrapNeedPushFlag finish abnormalTradeNoticeLogBo={}", bo);
        return bo;
    }

    /**
     * 监听交易对列表并且入库
     */
    private void monitorSymbolListAndStore(Set<String> symbolList) {
        log.info("AbnormalPriceServiceImpl.monitorSymbolListAndStore symbolList={}", symbolList);
        for (String symbol : symbolList) {
            worker.submit(() -> {
                HintManager hintManager = null;
                try {
                    // 判断是否有监控数据
                    StopWatch stopWatch = new StopWatch();
                    stopWatch.start();
                    AbnormalTradeNoticeLogBo abnormalTradeNoticeLogBo =
                            monitorAbnormalPriceEvent(symbol, splitQuoteAssetName(symbol));
                    stopWatch.stop();
                    log.info("AbnormalPriceServiceImpl.monitorAbnormalPriceEvent symbol={}, elapsedTime={}", symbol,
                            stopWatch.getTotalTimeSeconds());
                    if (null != abnormalTradeNoticeLogBo) {
                        List<AbnormalTradeNoticeLogBo> abnormalTradeNoticeLogBoList = Lists.newArrayList();
                        // 强制路由主库
                        hintManager = HintManager.getInstance();
                        hintManager.setMasterRouteOnly();
                        AbnormalTradeNoticeLogBo queryResult = abnormalTradeNoticeLogService
                                .findBySymbolAndBusinessTypeAndDescCode(abnormalTradeNoticeLogBo.getSymbol(),
                                        abnormalTradeNoticeLogBo.getBusinessType().name(),
                                        abnormalTradeNoticeLogBo.getDescCode().getDescCode());
                        log.info("AbnormalPriceServiceImpl.pushAbnormalTradeNotice symbol={}, queryResult={}", symbol,
                                queryResult);
                        if (null != queryResult) {
                            abnormalTradeNoticeLogBoList.add(queryResult);
                            String content =
                                    AbnormalPricePushMessageConvertor.wrapMessage(abnormalTradeNoticeLogBoList);
                            log.info("AbnormalPriceServiceImpl.pushAbnormalTradeNotice symbol={}, content={}", symbol,
                                    content);
                            abnormalTradePushNoticeService.pushWebMessage(AbnormalTradePushNoticeBusTypeEnum.APC, " ",
                                    content);
                        }
                    }
                } catch (Exception e) {
                    log.error("AbnormalPriceServiceImpl.monitorSymbolListAndStore Exception symbol=" + symbol, e);
                } finally {
                    if (null != hintManager) {
                        hintManager.close();
                    }
                }
            });
        }
    }

    /**
     * 判断是否是重复事件
     */
    private Boolean isDuplicateEventLog(AbnormalTradeNoticeLogBo before, AbnormalTradeNoticeLogBo after) {
        if (null == before) {
            return Boolean.FALSE;
        }
        Long diffTime = System.currentTimeMillis() - before.getCreatedTime().getTime();
        Long fiveMinute = 1000 * 60 * 5L;
        // 小于5钟，并且是已经重复推送过的文案了 我们认为是重复的
        // 一个文案5分钟只能触发一次
        if (diffTime.compareTo(fiveMinute) >= 0) {
            return Boolean.FALSE;
        }
        return Boolean.TRUE;
    }

    /**
     * 监控单个交易对的移动信息
     */
    private AbnormalTradeNoticeLogBo monitorAbnormalPriceEvent(String symbol, String quoteAssetName) {
        try {
            MathContext mc = new MathContext(8, RoundingMode.HALF_UP);
            List<KlineInfo> klineInfos = klineService.fetchKline(symbol, 5, IntervalEnum.ONE_MINUTES, null, null);
            if (CollectionUtils.isEmpty(klineInfos) || klineInfos.size() < 5) {
                log.info("AbnormalPriceServiceImpl.monitorAbnormalPriceEvent klineInfos empty symbol={} ", symbol);
                return null;
            }
            log.info("AbnormalPriceServiceImpl.monitorAbnormalPriceEvent symbol={}, size={}", symbol,
                    klineInfos.size());
            BigDecimal firstOpenPrice = klineInfos.get(0).getOpen();
            BigDecimal lastClosePrice = klineInfos.get(klineInfos.size() - 1).getClose();
            BigDecimal diffPrice = lastClosePrice.subtract(firstOpenPrice, mc);// 第一次与最后一次的差价
            BigDecimal absDiffPrice = diffPrice.abs(mc);// 差价绝对值
            BigDecimal priceChangePercent = diffPrice.divide(firstOpenPrice, mc);// 价格相对第一次的变动百分比
            AbnromalTradeNoticeDescEnum abnromalTradeNoticeDescEnum = convertDesc(priceChangePercent, symbol);
            log.info(
                    "AbnormalPriceServiceImpl.monitorAbnormalPriceEvent symbol={}, "
                            + "firstOpenPrice={},lastClosePrice={},"
                            + "diffPrice={},priceChangePercent={},abnromalTradeNoticeDescEnum={}",
                    symbol, firstOpenPrice, lastClosePrice, diffPrice, priceChangePercent, abnromalTradeNoticeDescEnum);
            // 这个条件命中代表这个币没有什么异常价格波动,或者价格差实在太小
            if (null == abnromalTradeNoticeDescEnum || needSkip(absDiffPrice, quoteAssetName)) {
                return null;
            }
            AbnormalTradeNoticeLogBo abnormalTradeNoticeLogBo = new AbnormalTradeNoticeLogBo();
            abnormalTradeNoticeLogBo.setSymbol(symbol);
            abnormalTradeNoticeLogBo.setQuoteAssetName(quoteAssetName);
            abnormalTradeNoticeLogBo.setDescCode(abnromalTradeNoticeDescEnum);
            abnormalTradeNoticeLogBo.setBusinessType(AbnromalTradeBusinessTypeEnum.AbnormalPrice);
            Map<String, String> data = Maps.newConcurrentMap();
            data.put("priceChangePercent", String.valueOf(priceChangePercent));
            abnormalTradeNoticeLogBo.setBusinessData(data);
            log.info("AbnormalPriceServiceImpl.monitorAbnormalPriceEvent symbol={}, AbnormalTradeNoticeLogBo={}",
                    abnormalTradeNoticeLogBo);
            // 尝试从数据库中获取历史记录
            AbnormalTradeNoticeLogBo before = abnormalTradeNoticeLogService.findBySymbolAndBusinessTypeAndDescCode(
                    symbol, AbnromalTradeBusinessTypeEnum.AbnormalPrice.name(),
                    abnormalTradeNoticeLogBo.getDescCode().getDescCode());
            Boolean isDuplicateEventFlag = isDuplicateEventLog(before, abnormalTradeNoticeLogBo);
            log.info("AbnormalPriceServiceImpl.monitorAbnormalPriceEvent symbol={} isDuplicateEventFlag={},before={}",
                    symbol, isDuplicateEventFlag, before);
            if (isDuplicateEventFlag) {
                return null;
            }
            abnormalTradeNoticeLogService.addOrUpdateAbnormalTradeNoticeLog(abnormalTradeNoticeLogBo);
            log.info(
                    "AbnormalPriceServiceImpl.monitorAbnormalPriceEvent symbol={} addOrUpdateAbnormalTradeNoticeLog={}",
                    symbol, abnormalTradeNoticeLogBo);
            return abnormalTradeNoticeLogBo;
        } catch (Exception e) {
            log.error("AbnormalPriceServiceImpl.monitorAbnormalPriceEvent Exception:", e);
            return null;
        }
    }

    private String splitQuoteAssetName(String symbol) {
        if (symbol.endsWith("BTC")) {
            return "BTC";
        }
        if (symbol.endsWith("USDT")) {
            return "USDT";
        }
        throw new IllegalArgumentException("illegal symbol=" + symbol);
    }

    /**
     * 某些归零的币实际上价格差只要小小的波动就会在涨幅上体现很多，这种币要剔除
     */
    private Boolean needSkip(BigDecimal absDiff, String quoteAssetName) {
        if ("BTC".equals(quoteAssetName) && absDiff.compareTo(new BigDecimal("0.00000004")) <= 0) {
            return Boolean.TRUE;
        }
        if ("USDT".equals(quoteAssetName) && absDiff.compareTo(new BigDecimal("0.0004")) <= 0) {
            return Boolean.TRUE;
        }
        return Boolean.FALSE;
    }

    /**
     * 获取配置
     */
    private AbnormalThresholdConfigModel getAbnormalThresholdConfig(String symbol) {
        Map<String, Object> param = Maps.newHashMap();
        param.put("symbol", symbol);
        param.put("businessType", AbnromalTradeBusinessTypeEnum.AbnormalPrice.name());
        AbnormalThresholdConfigModel abnormalThresholdConfigModel =
                abnormalThresholdConfigMapper.findBySymbolAndBusinessType(param);
        if (null == abnormalThresholdConfigModel) {
            param.put("symbol", "DEFAULT");
            abnormalThresholdConfigModel = abnormalThresholdConfigMapper.findBySymbolAndBusinessType(param);
        }
        log.info("AbnormalPriceServiceImpl.getAbnormalThresholdConfig symbol={}, abnormalThresholdConfigModel={}",
                symbol, abnormalThresholdConfigModel);
        return abnormalThresholdConfigModel;

    }

    private AbnromalTradeNoticeDescEnum convertDesc(BigDecimal priceChangePercent, String symbol) {
        AbnormalThresholdConfigModel abnormalThresholdConfigModel = getAbnormalThresholdConfig(symbol);
        if (abnormalThresholdConfigModel == null) {
            return null;
        }
        String[] priceRange = abnormalThresholdConfigModel.getBusinessStr().split(",");
        AbnromalTradeNoticeDescEnum abnromalTradeNoticeDescEnum = null;
        if (priceChangePercent.compareTo(new BigDecimal(priceRange[5])) > 0) {
            abnromalTradeNoticeDescEnum = AbnromalTradeNoticeDescEnum.AbnormalPrice_TO_THE_MOON;
        }
        if (priceChangePercent.compareTo(new BigDecimal(priceRange[4])) > 0
                && priceChangePercent.compareTo(new BigDecimal(priceRange[5])) <= 0) {
            abnromalTradeNoticeDescEnum = AbnromalTradeNoticeDescEnum.AbnormalPrice_BIG_PUMP;

        }
        if (priceChangePercent.compareTo(new BigDecimal(priceRange[3])) > 0
                && priceChangePercent.compareTo(new BigDecimal(priceRange[4])) <= 0) {
            abnromalTradeNoticeDescEnum = AbnromalTradeNoticeDescEnum.AbnormalPrice_Pump;

        }
        if (priceChangePercent.compareTo(new BigDecimal(priceRange[1])) > 0
                && priceChangePercent.compareTo(new BigDecimal(priceRange[2])) <= 0) {
            abnromalTradeNoticeDescEnum = AbnromalTradeNoticeDescEnum.AbnormalPrice_Dump;

        }
        if (priceChangePercent.compareTo(new BigDecimal(priceRange[0])) > 0
                && priceChangePercent.compareTo(new BigDecimal(priceRange[1])) <= 0) {
            abnromalTradeNoticeDescEnum = AbnromalTradeNoticeDescEnum.AbnormalPrice_BIG_Dump;

        }
        if (priceChangePercent.compareTo(new BigDecimal(priceRange[0])) <= 0) {
            abnromalTradeNoticeDescEnum = AbnromalTradeNoticeDescEnum.AbnormalPrice_MEGA_Dump;
        }
        log.info("AbnormalPriceServiceImpl.convertDesc symbol={},priceChangePercent={}, abnromalTradeNoticeDescEnum={}",
                symbol, priceChangePercent, abnromalTradeNoticeDescEnum);
        return abnromalTradeNoticeDescEnum;
    }

    /**
     * 获取风控开关配置
     */
    private Boolean getAbnormalPriceSwitch(String displayName) {
        String value = this.sysConfigService.getValue(displayName);
        if (null == value) {
            // 默认开启，就算没有配置开关
            return Boolean.FALSE;
        }
        if ("ON".equals(value)) {
            return Boolean.TRUE;
        }
        return Boolean.FALSE;
    }

    /**
     * 每个涨跌幅的数据都由不同的开关决定，具体的开关在admin上面，消息管理-》异常价格配置页面
     */
    private Boolean convertPushFlag(BigDecimal priceChangePercent, String symbol) {
        AbnormalThresholdConfigModel abnormalThresholdConfigModel = getAbnormalThresholdConfig(symbol);
        String[] priceRange = abnormalThresholdConfigModel.getBusinessStr().split(",");
        Boolean pushFlag = Boolean.FALSE;
        if (priceChangePercent.compareTo(new BigDecimal(priceRange[5])) > 0) {
            pushFlag = getAbnormalPriceSwitch("pump_to_the_moon_switch");
            return pushFlag;
        }
        if (priceChangePercent.compareTo(new BigDecimal(priceRange[4])) > 0
                && priceChangePercent.compareTo(new BigDecimal(priceRange[5])) <= 0) {
            pushFlag = getAbnormalPriceSwitch("big_pump_switch");
            return pushFlag;
        }
        if (priceChangePercent.compareTo(new BigDecimal(priceRange[3])) > 0
                && priceChangePercent.compareTo(new BigDecimal(priceRange[4])) <= 0) {
            pushFlag = getAbnormalPriceSwitch("pump_switch");
            return pushFlag;
        }
        if (priceChangePercent.compareTo(new BigDecimal(priceRange[1])) > 0
                && priceChangePercent.compareTo(new BigDecimal(priceRange[2])) <= 0) {
            pushFlag = getAbnormalPriceSwitch("dump_switch");
            return pushFlag;
        }
        if (priceChangePercent.compareTo(new BigDecimal(priceRange[0])) > 0
                && priceChangePercent.compareTo(new BigDecimal(priceRange[1])) <= 0) {
            pushFlag = getAbnormalPriceSwitch("big_dump_switch");
            return pushFlag;
        }
        if (priceChangePercent.compareTo(new BigDecimal(priceRange[0])) <= 0) {
            pushFlag = getAbnormalPriceSwitch("bump_to_the_hell_switch");
            return pushFlag;
        }
        log.info("AbnormalPriceServiceImpl.convertPushFlag symbol={},priceChangePercent={}", symbol,
                priceChangePercent);
        return pushFlag;
    }



    public List<AbnormalTradeNoticeInfo> getTestList() {
        Long nowTime = new Date().getTime();
        List<AbnormalTradeNoticeInfo> abnormalTradeNoticeInfos = Lists.newArrayList();
        AbnormalTradeNoticeInfo abnormalTradeNoticeInfo = new AbnormalTradeNoticeInfo();
        abnormalTradeNoticeInfo.setSymbol("BTCUSDT");
        abnormalTradeNoticeInfo.setQuoteAssetName("USDT");
        abnormalTradeNoticeInfo.setTime(nowTime);
        abnormalTradeNoticeInfo.setType("AbnormalPrice");
        Map<String, String> data = Maps.newConcurrentMap();
        data.put("priceChangePercent", String.valueOf("0.05"));
        data.put("descKey", "#{apc.tip4}");
        data.put("enDesc", "Pump");
        data.put("cnDesc", "拉升");
        data.put("descCode", "1");
        abnormalTradeNoticeInfo.setData(data);
        abnormalTradeNoticeInfos.add(abnormalTradeNoticeInfo);


        AbnormalTradeNoticeInfo abnormalTradeNoticeInfo1 = new AbnormalTradeNoticeInfo();
        abnormalTradeNoticeInfo1.setSymbol("BNBUSDT");
        abnormalTradeNoticeInfo1.setQuoteAssetName("USDT");
        abnormalTradeNoticeInfo1.setTime(nowTime + 100);
        abnormalTradeNoticeInfo1.setType("AbnormalPrice");
        Map<String, String> data1 = Maps.newConcurrentMap();
        data1.put("priceChangePercent", String.valueOf("0.25"));
        data1.put("descKey", "#{apc.tip4}");
        data1.put("enDesc", "To the MOON");
        data1.put("cnDesc", "火箭发射");
        data1.put("descCode", "-1");
        abnormalTradeNoticeInfo1.setData(data1);
        abnormalTradeNoticeInfos.add(abnormalTradeNoticeInfo1);

        Collections.sort(abnormalTradeNoticeInfos, new Comparator<AbnormalTradeNoticeInfo>() {
            @Override
            public int compare(AbnormalTradeNoticeInfo o1, AbnormalTradeNoticeInfo o2) {
                return o2.getTime().compareTo(o1.getTime());
            }
        });
        return abnormalTradeNoticeInfos;

    }
}
